//package com.autoai;
//
//
//import com.alibaba.fastjson.JSON;
//import com.autoai.mo.StandCarData;
//import com.autoai.mo.TripData;
//import org.apache.commons.lang3.time.FastDateFormat;
//import org.apache.flink.api.common.RuntimeExecutionMode;
//import org.apache.flink.api.common.functions.RichMapFunction;
//import org.apache.flink.api.common.serialization.SimpleStringSchema;
//import org.apache.flink.api.common.state.StateTtlConfig;
//import org.apache.flink.api.common.state.ValueState;
//import org.apache.flink.api.common.state.ValueStateDescriptor;
//import org.apache.flink.api.common.time.Time;
//import org.apache.flink.api.java.tuple.Tuple3;
//import org.apache.flink.configuration.Configuration;
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
//
//import java.util.Properties;
//
//public class flinkTrips {
//    public static void main(String[] args) throws Exception {
//        //TODO 0.env
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
//
//        //TODO 1.source
//        //准备kafka连接参数
////        Properties props  = new Properties();
////        props.setProperty("bootstrap.servers", "node1:9092");//集群地址
////        props.setProperty("group.id", "flink");//消费者组id
////        props.setProperty("auto.offset.reset","latest");//latest有offset记录从记录位置开始消费,没有记录从最新的/最后的消息开始消费 /earliest有offset记录从记录位置开始消费,没有记录从最早的/最开始的消息开始消费
////        props.setProperty("flink.partition-discovery.interval-millis","5000");//会开启一个后台线程每隔5s检测一下Kafka的分区情况,实现动态分区检测
////        props.setProperty("enable.auto.commit", "true");//自动提交(提交到默认主题,后续学习了Checkpoint后随着Checkpoint存储在Checkpoint和默认主题中)
////        props.setProperty("auto.commit.interval.ms", "2000");//自动提交的时间间隔
////        //使用连接参数创建FlinkKafkaConsumer/kafkaSource
////        FlinkKafkaConsumer<String> kafkaSource = new FlinkKafkaConsumer<String>("flink_kafka", new SimpleStringSchema(), props);
////
////        DataStream<StandCarData> kafkaDS = env.addSource(kafkaSource).map(s ->{
////            StandCarData data = JSON.parseObject(s, StandCarData.class);
////            return data;
////        });
//
//        DataStream<StandCarData> fileDS = env.readTextFile("D:\\02SoftWare\\IDEWorkSpace\\flink_study_47\\data\\input\\dir\\standcar.txt").map(s ->{
//            StandCarData data = JSON.parseObject(s, StandCarData.class);
//            System.out.println("---------"+data);
//            return data;
//        });
//
//        SingleOutputStreamOperator<TripData> valueState = fileDS.keyBy(s -> s.getVin()).map(new RichMapFunction<StandCarData, TripData>() {
//
//            private FastDateFormat df = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");
//            StateTtlConfig ttlConfig = StateTtlConfig
//                    .newBuilder(Time.seconds(600))
//                    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
//                    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
//                    .build();
//            private ValueState<TripData> valueState;
//
//            @Override
//            public void open(Configuration parameters) throws Exception {
//                //创建状态描述器
//                ValueStateDescriptor stateDescriptor = new ValueStateDescriptor("valueState", TripData.class);
////                stateDescriptor.enableTimeToLive(ttlConfig);
//                //根据状态描述器获取/初始化状态
//                valueState = getRuntimeContext().getState(stateDescriptor);
//            }
//
//            @Override
//            public TripData map(StandCarData value) throws Exception {
//                if (System.currentTimeMillis()-value.getTimeStamp() > 1000*60*10000){
//                    System.out.println("-----------"+value);
//                    valueState.clear();
//                    return valueState.value();
//                }else{
//                    //获取状态
//                    TripData historyValue = valueState.value();
//                    System.out.println("-----------"+value);
//                    long currentTimeMillis = System.currentTimeMillis();
//                    String dateTime = df.format(currentTimeMillis);
//                    //判断状态
//                    System.out.println(valueState.value());
//                    if (historyValue == null) {
//                        historyValue = new TripData(
//                                value.getVin() + "_" + value.getUserId() + "_" + value.getTimeStamp(),
//                                value.getVin(),
//                                value.getUserId(),
//                                value.getMileage(),
//                                0.0,
//                                0L,
//                                value.getTimeStamp(),
//                                value.getTimeStamp(),
//                                value.getLon(),
//                                value.getLat(),
//                                value.getLon(),
//                                value.getLat(),
//                                dateTime
//                        );
//                        //更新状态
//                        valueState.update(historyValue);
//
//                        return valueState.value();
//                    } else {
//                        Double diff = value.getMileage() - historyValue.getT_end_mil();
//                        historyValue = new TripData(
//                                value.getVin() + "_" + value.getUserId() + "_" + value.getTimeStamp(),
//                                value.getVin(),
//                                value.getUserId(),
//                                value.getMileage(),
//                                diff,
//                                0L,
//                                value.getTimeStamp(),
//                                value.getTimeStamp(),
//                                value.getLon(),
//                                value.getLat(),
//                                value.getLon(),
//                                value.getLat(),
//                                dateTime
//                        );
//                        //更新状态
//                        valueState.update(historyValue);
//
//                        return valueState.value();
//                    }
//                }
//
//            }
//        });
//
//        env.execute();
//    }
//
//}
